package com.sun.electric.compression.api;

import java.io.IOException;
import java.io.OutputStream;
import java.util.LinkedList;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;

import com.sun.electric.compression.bzip2.BZip2Constants;
import com.sun.electric.compression.bzip2.BZip2OutputStream;

/**
 * @see BZip2OutputStream
 * @author Daniel.Schmidt
 */
public class BZIP2OutputStreamJava extends BZip2OutputStream {

	/* The threadpool with nThreads threads waiting to execute tasks */
	private ExecutorService threadpool;
	/* All tasks submitted to the threadpool are queued */
	private LinkedList<Future<BZIP2BlockCompressorJava>> writingQueue;
	/* Amount of tasks submitted */
	private int amountOfTasks = 0;
	/* The current compressor whose data shall be written to the file at next */
	private Future<BZIP2BlockCompressorJava> currentCompressor = null;
	/* Amount of threads */
	private int nThreads = 0;

	/**
	 * C'tor Initializes the threadpool, the writing queue and the
	 * BlockCompressor for the first block of data.
	 * @throws IOException 
	 */
	public BZIP2OutputStreamJava(OutputStream outputStream,
			int blockSizeFactor, int nThreads) throws IOException {
		super(outputStream, blockSizeFactor);
		if (nThreads > 1)
			// Initialize the thread pool with n - 1 thread
			// Notice: the main thread does work himself!
			threadpool = Executors.newFixedThreadPool(nThreads - 1);
		this.nThreads = nThreads;
		writingQueue = new LinkedList<Future<BZIP2BlockCompressorJava>>();

		initialiseNextBlock();
	}

	/**
	 * Initializes the next BlockCompressor containing the current block of
	 * data.
	 */
	@Override
	public void initialiseNextBlock() {

		this.blockCompressor = new BZIP2BlockCompressorJava(
				this.bitOutputStream, this.streamBlockSize);
		amountOfTasks = amountOfTasks + 1;
	}

	/**
	 * If the current compressor is full the following alternatives are
	 * possible:
	 * 
	 * If there are "too much" elements waiting in the writing queue: - Check if
	 * the current compressor has finished -> If yes, write the compressed data
	 * to the file -> If no, compress a block yourself and check later again If
	 * the writing queue has an acceptable size, submit the procession of the
	 * current block and execute it in parallel with help of the thread pool.
	 * The returned future indicates the completion of the task and is stored in
	 * the writing queue.
	 * 
	 * Notice: This handling is needed to prevent a java out of memory
	 * exception. This exception appears when all compressors are created
	 * initially. Now the main thread has a threshold that determines the number
	 * of compressors in the queue. If the threshold is exceeded, it tries to
	 * write finished data at first.
	 * @throws IOException
	 */
	@Override
	public void closeBlock() throws IOException {

		// Set the current compressor if it is still null. Therefore the writing
		// queue
		// has to contain an element.
		if (currentCompressor == null && !writingQueue.isEmpty()) {
			currentCompressor = writingQueue.poll();
		}

		boolean closed = false;
		do {
			if (nThreads == 1 || writingQueue.size() > ((nThreads - 1) * 2)) {
				// Only check if current compressor is done, otherwise compress
				// yourself during
				// the meantime
				if (currentCompressor != null && currentCompressor.isDone()) {
					writeToStream();
				} else {
					BZIP2BlockCompressorJava comp = ((BZIP2BlockCompressorJava) this.blockCompressor)
							.call();
					writingQueue.add(new FutureOfCompressor(comp));
					closed = true;
				}
			} else {
				Future<BZIP2BlockCompressorJava> f = threadpool
						.submit((BZIP2BlockCompressorJava) this.blockCompressor);
				writingQueue.add(f);
				closed = true;
			}
		} while (!closed);
	}

	/**
	 * After all blocks are compressed in parallel, the results can be written
	 * out to the file. Finally the thread pool can be shut down.
	 * @throws IOException
	 */
	@Override
	public void finish() throws IOException {

		if (!this.streamFinished) {
			this.streamFinished = true;
			try {

				closeBlock();
				while (amountOfTasks > 0) {
					if (currentCompressor == null) {
						currentCompressor = writingQueue.poll();
					}
					writeToStream();
				}

				this.bitOutputStream.writeBits(24,
						BZip2Constants.STREAM_END_MARKER_1);
				this.bitOutputStream.writeBits(24,
						BZip2Constants.STREAM_END_MARKER_2);
				this.bitOutputStream.writeInteger(this.streamCRC);
				this.bitOutputStream.flush();
				this.outputStream.flush();
			} finally {
				if (nThreads > 1)
					threadpool.shutdownNow();
				this.blockCompressor = null;
			}
		}
	}

	/**
	 * Writes the information(header, compressed data, CRC) of the finished
	 * compressor to the file. Due to the queue it is ensured that the order of
	 * the compressed block data is maintained and written out correctly.
	 * @throws IOException
	 */
	private void writeToStream() throws IOException {

		BZIP2BlockCompressorJava compressor = null;
		try {
			compressor = currentCompressor.get();
		} catch (InterruptedException e) {
			e.printStackTrace();
		} catch (ExecutionException e) {
			e.printStackTrace();
		}

		amountOfTasks = amountOfTasks - 1;

		// Write out the block header
		this.bitOutputStream
				.writeBits(24, BZip2Constants.BLOCK_HEADER_MARKER_1);
		this.bitOutputStream
				.writeBits(24, BZip2Constants.BLOCK_HEADER_MARKER_2);
		this.bitOutputStream.writeInteger(compressor.getHeaderCRC());
		this.bitOutputStream.writeBoolean(false); // Randomised block flag. We
													// never create randomised
													// blocks
		this.bitOutputStream.writeBits(24, compressor.getBwtStartPointer());

		// Write out the symbol map
		compressor.writeSymbolMap();

		compressor.getEncoder().writeEncoding();

		int blockCRC = compressor.getCRC();
		this.streamCRC = ((this.streamCRC << 1) | (this.streamCRC >>> 31))
				^ blockCRC;

		// Set new compressor
		currentCompressor = writingQueue.poll();
	}

	/**
	 * Holds an BZIP2BlockCompressorScala and implements the future interface.
	 * This class is needed for the main thread to process a block itself and
	 * store the result as future in the general writing queue.
	 */
	private class FutureOfCompressor implements
			Future<BZIP2BlockCompressorJava> {

		BZIP2BlockCompressorJava compressor;

		private FutureOfCompressor(BZIP2BlockCompressorJava compressor) {
			this.compressor = compressor;
		}

		public boolean cancel(boolean mayInterruptIfRunning) {
			return false;
		}

		public BZIP2BlockCompressorJava get() {
			return compressor;
		}

		public BZIP2BlockCompressorJava get(long timeout, TimeUnit unit) {
			return compressor;
		}

		public boolean isCancelled() {
			return false;
		}

		public boolean isDone() {
			return true;
		}
	}
}
